package rabbitMqLearn.mq.work_model.client;

import com.rabbitmq.client.*;
import rabbitMqLearn.mq.config.Config;

import java.io.IOException;
import java.util.Collections;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeoutException;

/**
 * RPC 通信
 * 使用俩个队列来储存请求和响应
 * 客户端逻辑是 将消息发送给请求队列，并从响应队列中拿取到响应
 * 服务端逻辑是 从请求队列中拿到消息，并且将响应发送给响应队列
 * 所以客户端服务端都需要实现生产者和服务者逻辑
 *
 * 确认设置
 *
 *
 */
public class RPCClient {
    Integer MESSAGE_COUNT = 10;

    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        RPCClient client = new RPCClient();
//        client.rpc();
//        client.singleConfirm();
//        client.batch();
        client.asynchronous();
    }

    void rpc() throws IOException, TimeoutException, InterruptedException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(Config.HOST);
        factory.setPort(Config.PORT);
        factory.setVirtualHost(Config.VIRTUALHOST);
        factory.setUsername(Config.USERNAME);
        factory.setPassword(Config.PASSWORD);

        Connection connection = factory.newConnection();

        Channel channel = connection.createChannel();


        // 请求的标识
        String id = UUID.randomUUID().toString();

        //设置请求的属性
        AMQP.BasicProperties build = new AMQP.BasicProperties.Builder()
                .correlationId(id)// 设置请求的id
                .replyTo(Config.RPC_RESPONSE_QUEUE) //设置响应放到那个队列中
                .build();

        String msg = "rpc hello";

        //发送消息
        for (int i = 0; i < 10; i++) {
            channel.basicPublish("", Config.RPC_REQUEST_QUEUE, build, msg.getBytes());
        }

        //接收消息
        //需要一个阻塞队列来等待响应
        BlockingQueue<String> response = new ArrayBlockingQueue<>(1);

        //使用consume来获取到响应队列中的响应
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String message = new String(body);
                if (id.equals(properties.getCorrelationId())) {
                    response.offer(message);
                }
            }
        };

        //消费response中的消息
        while (true) {
            channel.basicConsume(Config.RPC_RESPONSE_QUEUE, true, consumer);
            System.out.println("收到响应: " + response.take());
        }
    }

    /**
     * Publish Confirm 发布确认
     * 作为消息中间件可能会面临消息丢失的情况，而对于消息的丢失主要原因有三种
     * 1. 生产者网络问题发送给mq失败
     * 2. mq储存的消息丢失
     * 3. 消费者接收到消息失败
     * 问题的解决 1 发布确认机制解决   2 消息持久化  3 消息应答机制
     */

    /**
     * 发布确认有三种确认机制
     * 1 单独确认
     * 2 批量确认
     * 3 异步确认
     */

    /**
     * 单独确认
     */
    void singleConfirm(){
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(Config.HOST);
        factory.setPort(Config.PORT);
        factory.setVirtualHost(Config.VIRTUALHOST);
        factory.setUsername(Config.USERNAME);
        factory.setPassword(Config.PASSWORD);

        try(Connection connection = factory.newConnection()){
            Channel channel = connection.createChannel();
            //开启信道确认模式
            channel.confirmSelect();

            //声明队列
            channel.queueDeclare(Config.RPC_REQUEST_CONFIRM_QUEUE, true, false, false, null);

            //发送消息
            String s = "";
            long startTime = System.currentTimeMillis();
            for(int i = 0; i < MESSAGE_COUNT; i++){
                s = "发送的消息: " + i;

                channel.basicPublish("", Config.RPC_REQUEST_CONFIRM_QUEUE, null, s.getBytes());

                //等待确认
                channel.waitForConfirms(5000);
            }
            long endTime = System.currentTimeMillis();
            System.out.println("单独确认消耗时间: " + (endTime - startTime) / 1000.0);
        } catch (IOException e) {
            throw new RuntimeException(e);
        } catch (TimeoutException e) {
            throw new RuntimeException(e);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 批量确认
     */
    void batch(){
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(Config.HOST);
        factory.setPort(Config.PORT);
        factory.setVirtualHost(Config.VIRTUALHOST);
        factory.setUsername(Config.USERNAME);
        factory.setPassword(Config.PASSWORD);
        try(Connection connection = factory.newConnection()){
            Channel channel = connection.createChannel();
            //开启信道确认模式
            channel.confirmSelect();

            //声明队列
            channel.queueDeclare(Config.RPC_REQUEST_CONFIRM_QUEUE, true, false, false, null);

            //发送消息
            String s = "";

            int batchSize = 1000;
            int outSize = 0;

            long startTime = System.currentTimeMillis();
            for(int i = 0; i < MESSAGE_COUNT; i++){
                s = "发送的消息: " + i;

                channel.basicPublish("", Config.RPC_REQUEST_CONFIRM_QUEUE, null, s.getBytes());

                outSize++;

                //每100 条确认一次
                if (outSize == batchSize) {
                    channel.waitForConfirmsOrDie(5000);
                    outSize = 0;
                }
            }
            long endTime = System.currentTimeMillis();
            System.out.println("批量确认消耗时间: " + (endTime - startTime) / 1000.0);
        } catch (IOException e) {
            throw new RuntimeException(e);
        } catch (TimeoutException e) {
            throw new RuntimeException(e);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }


    //异步确认
    void asynchronous() throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(Config.HOST);
        factory.setPort(Config.PORT);
        factory.setVirtualHost(Config.VIRTUALHOST);
        factory.setUsername(Config.USERNAME);
        factory.setPassword(Config.PASSWORD);

        try (Connection connection = factory.newConnection();){
            Channel channel = connection.createChannel();
            //开启信道确认模式
            channel.confirmSelect();

            //声明队列
            channel.queueDeclare(Config.RPC_REQUEST_CONFIRM_QUEUE, true, false, false, null);

            //储存的是有序的确认id
            SortedSet<Long> confirmSet = Collections.synchronizedSortedSet(new TreeSet<>());

            channel.addConfirmListener(new ConfirmListener() {
                //收到deliveryTag消息的处理
                @Override
                public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                    try {
                        //是否批量确认
                        if (multiple) {
                            System.out.println("批量确认");
                            //deliveryTag 从这个开始序号之前的都被收到了
                            confirmSet.headSet(deliveryTag + 1).clear();
                        } else {
                            System.out.println("单独确认");

                            channel.waitForConfirms();
                        }
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                }

                //未收到deliveryTag消息的处理
                @Override
                public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                    System.out.println("未收到 deliveryTag：" + deliveryTag);
                    try {
                        //是否批量确认
                        if (multiple) {
                            //deliveryTag 从这个开始序号之前的都被收到了
                            confirmSet.headSet(deliveryTag + 1).clear();
                        } else {
                            channel.waitForConfirms();
                        }
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                }
            });

            //发送消息
            String s = "";
            long startTime = System.currentTimeMillis();
            for (int i = 0; i < MESSAGE_COUNT; i++) {
                s = "发送的消息: " + i;
                long nextPublishSeqNo = channel.getNextPublishSeqNo();
                channel.basicPublish("", Config.RPC_REQUEST_CONFIRM_QUEUE, null, s.getBytes());
                confirmSet.add(nextPublishSeqNo);
            }

            if (!confirmSet.isEmpty()) {
                Thread.sleep(10);
            }

            long endTime = System.currentTimeMillis();
            System.out.println("消耗时间: " + (endTime - startTime) / 1000.0);
        } catch (IOException | InterruptedException e) {
            System.out.println(e.getMessage());
            throw new RuntimeException(e);
        }
    }
}
